/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月22日 上午10:58:08
*/
package cn.com.guage.flink.transformation;

import java.util.Arrays;

import org.apache.flink.api.common.functions.MapFunction;
/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月22日 上午10:58:08
*/
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.runtime.partitioner.GlobalPartitioner;

import com.alibaba.fastjson.JSON;

import cn.com.guage.flink.domain.UserAction;

/**
 * Summary:
 * Reduce: 基于ReduceFunction进行滚动聚合，并向下游算子输出每次滚动聚合后的结果。
 */
public class DataStreamReduceOperator {
    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        
        // 输入: 用户行为。某个用户在某个时刻点击或浏览了某个商品，以及商品的价格。
        DataStreamSource<UserAction> source = env.fromCollection(Arrays.asList(
                new UserAction("userID1", 1293984000, "click", "productID1", 10),
                new UserAction("userID2", 1293984001, "browse", "productID2", 8),
                new UserAction("userID2", 1293984002, "browse", "productID2", 8),
                new UserAction("userID2", 1293984003, "browse", "productID2", 8),
                new UserAction("userID1", 1293984002, "click", "productID1", 10),
                new UserAction("userID1", 1293984003, "click", "productID3", 10),
                new UserAction("userID1", 1293984004, "click", "productID1", 10)
        ));
        // 转换: KeyBy对数据重分区
        KeyedStream<UserAction, String> keyedStream = source.keyBy(new KeySelector<UserAction, String>() {
            /**
			 * 
			 */
			private static final long serialVersionUID = -8071512466242081495L;

			@Override
            public String getKey(UserAction value) throws Exception {
                return value.getUserId();
            }
        });
       
        keyedStream.map(new MapFunction<UserAction, String>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 6781172680960365561L;

			@Override
			public String map(UserAction value) throws Exception {
				// TODO Auto-generated method stub
				return "keyby后结果为：" + JSON.toJSONString(value);
			}
		}).print();
       // keyedStream.print();
        // 转换: Reduce滚动聚合。这里,滚动聚合每个用户对应的商品总价格。
		SingleOutputStreamOperator<String> result = keyedStream.reduce(new ReduceFunction<UserAction>() {
            /**
			 * 
			 */
			private static final long serialVersionUID = 5866793996646848614L;

			@Override
            public UserAction reduce(UserAction value1, UserAction value2) throws Exception {
                int newProductPrice = value1.getProductPrice() + value2.getProductPrice();
                return new UserAction(value1.getUserId(), -1, "", "", newProductPrice);
            }
        }).map(new MapFunction<UserAction, String>() {

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public String map(UserAction value) throws Exception {
				return JSON.toJSONString(value);
			}
		});

        // 输出: 将每次滚动聚合后的结果输出到控制台。
        //3> UserAction(userID=userID2, eventTime=1293984001, eventType=browse, productID=productID2, productPrice=8)
        //3> UserAction(userID=userID2, eventTime=-1, eventType=, productID=, productPrice=16)
        //3> UserAction(userID=userID2, eventTime=-1, eventType=, productID=, productPrice=24)
        //4> UserAction(userID=userID1, eventTime=1293984000, eventType=click, productID=productID1, productPrice=10)
        //4> UserAction(userID=userID1, eventTime=-1, eventType=, productID=, productPrice=20)
        //4> UserAction(userID=userID1, eventTime=-1, eventType=, productID=, productPrice=30)
        //4> UserAction(userID=userID1, eventTime=-1, eventType=, productID=, productPrice=40)
        result.print();

        env.execute();
    }
}
